package org.example.app.dwd;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.example.utils.DruidDSUtil;
import org.example.utils.DimUtil;
import org.example.utils.MyKafkaUtil;
import java.sql.Connection;
import java.util.Collections;
import java.util.concurrent.TimeUnit;

/**
 * 内容事实表处理器
 */
public class DwdMarketContentProcessor {
    
    public static void main(String[] args) throws Exception {
        // 创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        // 从Kafka读取数据
        String topic = "ods_market_content";
        String groupId = "dwd_market_content_group";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        
        // 添加水位线
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(jsonStr -> {
            JSONObject jsonObject = JSONObject.parseObject(jsonStr);
            return jsonObject;
        }).assignTimestampsAndWatermarks(
            WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts");
                    }
                })
        );
        
        // 异步关联店铺维度
        SingleOutputStreamOperator<JSONObject> withShopDS = AsyncDataStream.unorderedWait(
            jsonObjDS,
            new AsyncFunction<JSONObject, JSONObject>() {
                @Override
                public void asyncInvoke(JSONObject content, ResultFuture<JSONObject> resultFuture) throws Exception {
                    String shopId = content.getString("shop_id");
                    
                    try {
                        Connection connection = getConnection();
                        JSONObject shopDim = DimUtil.getDimInfo(connection, "dim_shop_info", shopId);
                        connection.close();
                        
                        if (shopDim != null) {
                            content.put("shop_name", shopDim.getString("SHOP_NAME"));
                            content.put("shop_type", shopDim.getString("SHOP_TYPE"));
                        }
                        
                        resultFuture.complete(Collections.singleton(content));
                    } catch (Exception e) {
                        e.printStackTrace();
                        resultFuture.complete(Collections.singleton(content));
                    }
                }
            },
            60, TimeUnit.SECONDS
        );
        
        // 将结果写入Kafka
        withShopDS.map(new MapFunction<JSONObject, String>() {
            @Override
            public String map(JSONObject jsonObject) throws Exception {
                return jsonObject.toJSONString();
            }
        }).addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_market_content"));
        
        // 执行任务
        env.execute("Content Fact Table Processing Job");
    }
    
    // 用于获取数据库连接的工具方法
    private static Connection getConnection() throws Exception {
        return DruidDSUtil.createDataSource().getConnection();
    }
}